package com.itheima.es;

import com.alibaba.fastjson.JSONObject;
import org.apache.http.HttpHost;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;

import java.sql.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Author : coderzhao89
 * @Description :   把数据库中sku表的数据导入到ES中 ，导入哪些数据：id，name，price，image，
 * categoryName，brandName，createTime，spuId,saleNum,commentNum
 * @Date Created in  9:16 2019/7/10
 * @Modified By :
 * @Company www.itheima.com
 */
public class Db2ES {

    public static void main(String[] args)  throws  Exception{
        List<Map> data = getData();
        db2Es(data);
    }

    /**
     * 从数据库中获取sku数据
     */
    public static  List<Map>   getData() throws  Exception{
        // JDBC 操作数据库

        // 1、注册驱动
        Class.forName("com.mysql.jdbc.Driver");
        // 2、获取连接
        Connection connection = DriverManager.getConnection("jdbc:mysql://127.0.0.1:3306/qingcheng_goods?characterEncoding=UTF8", "root", "root");
        String  sql="SELECT  " +
                "   sku.id as  id,  " +
                "   sku.`name`  as name ,  " +
                "   sku.`image`  as image ,  " +
                "   sku.brand_name  as brandName,  " +
                "   sku.price as price,  " +
                "   sku.create_time as createTime,  " +
                "   sku.sale_num as saleNum,  " +
                "   sku.comment_num  as commentNum,  " +
                "   sku.category_name as categoryName,  " +
                "   sku.spu_id as spuId,  " +
                "   sku.spec as spec  " +
                "FROM    " +
                " tb_sku  as  sku   " +
                "WHERE  " +
                " `status`= 1"+"" +
                " limit 0 , 100";
        // 3、得到一个发送和执行sql对象
        PreparedStatement preparedStatement = connection.prepareStatement(sql);
        ResultSet resultSet = preparedStatement.executeQuery();
        // 4、遍历结果集

        List <Map>  skuList = new ArrayList<Map>();
        while (resultSet.next()){
            Map sku  =  new HashMap();
            Map specMap = new HashMap();
            String id = resultSet.getString("id");
            int price = resultSet.getInt("price");
            String image = resultSet.getString("image");
            String name = resultSet.getString("name");
            int saleNum = resultSet.getInt("saleNum");
            int commentNum = resultSet.getInt("commentNum");
            String spec = resultSet.getString("spec");
            String spuId = resultSet.getString("spuId");
            Date sqldate = resultSet.getDate("createTime");
            String brandName = resultSet.getString("brandName");
            String categoryName = resultSet.getString("categoryName");
            // ES 要的是一个{ }  对象
            specMap= JSONObject.parseObject(spec, Map.class);
            // 日期转换： sql 的Date 转成 util的Date
            java.util.Date  createTime = new java.util.Date( sqldate.getTime());
            /**
             *  封装数据
             */

            sku.put("id",id);
            sku.put("name",name);
            sku.put("price",price);
            sku.put("brandName",brandName);
            sku.put("categoryName",categoryName);
            sku.put("saleNum",saleNum);
            sku.put("spec",specMap);
            sku.put("image",image);
            sku.put("commentNum",commentNum);
            sku.put("spuId",spuId);
            sku.put("createTime",createTime);

            skuList.add(sku);
        }

        // 测试
        for (Map map : skuList ){
            System.out.println(map);
            }
        System.out.println(skuList.size());

        resultSet.close();
        preparedStatement.close();
        connection.close();
        return  skuList;
    }

    /**
     *   把数据导入 ES 中
     * @param data
     */
    public  static  void  db2Es(List<Map> data) throws  Exception{

        HttpHost http=new HttpHost("127.0.0.1",9200,"http");
        RestClientBuilder builder= RestClient.builder(http);//rest构建器
        RestHighLevelClient restHighLevelClient=new RestHighLevelClient(builder);//高级客户端对象

        BulkRequest bulkRequest=new BulkRequest();
        for (Map map :data){
          String id=  (String)map.get("id");
            IndexRequest indexRequest=new IndexRequest("sku","doc",id);
            map.remove("id");
            indexRequest.source(map);
            bulkRequest.add(indexRequest);
        }

        BulkResponse response = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
        int status = response.status().getStatus();
        System.out.println(status);

        String message = response.buildFailureMessage();
        System.out.println(message);
        restHighLevelClient.close();
    }

}
